package com.cloudsubassemblykafka10024.tranactions;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.*;

/**
 * @Description: Kafka 事务
 * @Author zx
 * @Date 2024/12/24 19:55
 * @Version 1.0
 */
public class KafkaProducerAndConsumerTransactions_001 {
    public static void main(String[] args) {

        KafkaProducer<String, String> properties = buildKafkaProducer();
        KafkaConsumer<String, String> consumer = buildKafkaConsumer();

        // 开启事务 初始化事务
        properties.initTransactions();

        consumer.subscribe(Arrays.asList("messageDML005"));

        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            if (!consumerRecords.isEmpty()) {
                Iterator<ConsumerRecord<String, String>> iterator = consumerRecords.iterator();
                Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();
                // 开始事务
                properties.beginTransaction();
                try {
                    // 迭代数据 进行业务处理
                    while (iterator.hasNext()) {
                        ConsumerRecord<String, String> record = iterator.next();
                        // 记录消费分区的偏移量元数据
                        offsetMap.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1));

                        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("messageDML006", record.key(),
                                record.value() + ":ZhangXinTest");
                        properties.send(producerRecord);
                    }
                    // 提交事务 提交偏移量
                    properties.sendOffsetsToTransaction(offsetMap, "group1");
                    // 提交事务 生产者
                    properties.commitTransaction();

                } catch (Exception e) {
                    System.out.println("事务提交失败" + e.getMessage());
                    // 终止事务
                    properties.abortTransaction();
                }

            }
        }
    }

    public static KafkaProducer<String, String> buildKafkaProducer() {
        // 创建 Kafka Producer
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.18.220:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 设置事务ID
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction-id-" + UUID.randomUUID());

        // 配置Kafka的批处理大小
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 1024);
        // 配置Kafka的延迟时间 5ms  如果batch中的数据不足1024 则立即发送
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 5);

        // 配置Kafka的重试和幂等机制
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 20000);

        return new KafkaProducer<>(properties);
    }

    public static KafkaConsumer<String, String> buildKafkaConsumer() {

        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.18.220:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group6");
        // 设置消费者的消费事务的隔离级别
        properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        // 必须关闭消费者端的 offset 提交功能
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);


        return new KafkaConsumer<>(properties);
    }

}
